package com.sample.springbatch.job.demo.jobtransactional;

import com.sample.springbatch.config.BaseConfig;
import com.sample.springbatch.job.demo.user.UserJpaPagingItemReader;
import com.sample.springbatch.model.User;
import com.sample.springbatch.model.UserLog;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.Map;

/**
 * 异步线程需要修改主线程未提交的数据(包括新增和更新)，需要等待主线程数据提交后，异步线程才能提交。
 *
 * @author zwg
 * @date 2019-05-11 10:59
 **/

@Configuration
public class JobTransactionalConfig extends BaseConfig {

    Logger logger = LoggerFactory.getLogger(JobTransactionalConfig.class);

    @Autowired
    UserService userService;

    @Bean(name = "jobTransactionalJob")
    public Job importUserJob(@Qualifier("jobTransactionalJob_stepOne") Step stepOne) {
        return jobBuilderFactory.get("jobTransactionalJob")
                .incrementer(new RunIdIncrementer())
                .flow(stepOne)
                .end()
                .build();
    }

    @Bean(name = "jobTransactionalJob_stepOne")
    public Step stepOne(@Qualifier("jobTran_userJpaPagingItemReader") UserJpaPagingItemReader userJpaPagingItemReader,
                        @Qualifier("jobTran_processor") ItemProcessor processor,
                        @Qualifier("jobTran_userItemWriter") ItemWriter userItemWriter) {
        return stepBuilderFactory.get("stepOne")
                .<User, User> chunk(20)
                .reader(userJpaPagingItemReader)
                .processor(processor)
                .writer(userItemWriter)
                .build();
    }

    @Bean(name = "jobTran_userJpaPagingItemReader")
    @StepScope
    public UserJpaPagingItemReader<User> userJpaPagingItemReader(@Value("#{jobParameters}") Map params){
        return new UserJpaPagingItemReader<User>(params);
    }

    @Bean(name="jobTran_userItemWriter")
    public ItemWriter<User> userItemWriter() {
        return new ItemWriter<User>() {
            @Override
            public void write(List<? extends User> items) throws Exception {
                Thread.currentThread().sleep(15000);
                //throw new Exception("test");
            }

        };
    }

    @Bean(name = "jobTran_processor")
    public ItemProcessor<User,User> processor() {
        return new ItemProcessor<User,User>() {

            @Override
            public User process(User item) throws Exception {
                //item.setAge(item.getAge()+1);
                logger.info("user:{}",item);
                UserLog log = userService.updateUserAge(item);
                return item;
            }
        };
    }


}
